/**
 * Software License, Version 1.0
 * 
 * Copyright 2003 The Trustees of Indiana University.  All rights reserved.
 * 
 *
 *Redistribution and use in source and binary forms, with or without 
 *modification, are permitted provided that the following conditions are met:
 *
 *1) All redistributions of source code must retain the above copyright notice,
 * the list of authors in the original source code, this list of conditions and
 * the disclaimer listed in this license;
 *2) All redistributions in binary form must reproduce the above copyright 
 * notice, this list of conditions and the disclaimer listed in this license in
 * the documentation and/or other materials provided with the distribution;
 *3) Any documentation included with all redistributions must include the 
 * following acknowledgement:
 *
 *"This product includes software developed by the Community Grids Lab. For 
 * further information contact the Community Grids Lab at 
 * http://communitygrids.iu.edu/."
 *
 * Alternatively, this acknowledgement may appear in the software itself, and 
 * wherever such third-party acknowledgments normally appear.
 * 
 *4) The name Indiana University or Community Grids Lab or NaradaBrokering, 
 * shall not be used to endorse or promote products derived from this software 
 * without prior written permission from Indiana University.  For written 
 * permission, please contact the Advanced Research and Technology Institute 
 * ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202.
 *5) Products derived from this software may not be called NaradaBrokering, 
 * nor may Indiana University or Community Grids Lab or NaradaBrokering appear
 * in their name, without prior written permission of ARTI.
 * 
 *
 * Indiana University provides no reassurances that the source code provided 
 * does not infringe the patent or any other intellectual property rights of 
 * any other entity.  Indiana University disclaims any liability to any 
 * recipient for claims brought by any other entity based on infringement of 
 * intellectual property rights or otherwise.  
 *
 *LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH NO 
 *WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY GIVES
 *NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF 
 *INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS. 
 *INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS", 
 *"VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.  
 *LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR 
 *ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION 
 *GENERATED USING SOFTWARE.
 */
/**
 * Software License, Version 1.0 Copyright 2003 The Trustees of Indiana
 * University. All rights reserved. Redistribution and use in source and binary
 * forms, with or without modification, are permitted provided that the
 * following conditions are met: 1) All redistributions of source code must
 * retain the above copyright notice, the list of authors in the original source
 * code, this list of conditions and the disclaimer listed in this license; 2)
 * All redistributions in binary form must reproduce the above copyright notice,
 * this list of conditions and the disclaimer listed in this license in the
 * documentation and/or other materials provided with the distribution; 3) Any
 * documentation included with all redistributions must include the following
 * acknowledgement: "This product includes software developed by the Community
 * Grids Lab. For further information contact the Community Grids Lab at
 * http://communitygrids.iu.edu/." Alternatively, this acknowledgement may
 * appear in the software itself, and wherever such third-party acknowledgments
 * normally appear. 4) The name Indiana University or Community Grids Lab or
 * NaradaBrokering, shall not be used to endorse or promote products derived
 * from this software without prior written permission from Indiana University.
 * For written permission, please contact the Advanced Research and Technology
 * Institute ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202. 5)
 * Products derived from this software may not be called NaradaBrokering, nor
 * may Indiana University or Community Grids Lab or NaradaBrokering appear in
 * their name, without prior written permission of ARTI. Indiana University
 * provides no reassurances that the source code provided does not infringe the
 * patent or any other intellectual property rights of any other entity. Indiana
 * University disclaims any liability to any recipient for claims brought by any
 * other entity based on infringement of intellectual property rights or
 * otherwise. LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH
 * NO WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY
 * GIVES NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF
 * INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS.
 * INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS",
 * "VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.
 * LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR
 * ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION
 * GENERATED USING SOFTWARE.
 */
package cgl.narada.transport.https;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.Socket;
import java.net.SocketException;

import org.apache.log4j.Logger;

import cgl.narada.transport.TransportException;
import cgl.narada.transport.proxy.ProxyTunnel;
import cgl.narada.transport.sslHttpBase.LinkMessageKeepAlive;
import cgl.narada.transport.sslHttpBase.Transport;
import cgl.narada.transport.sslHttpBase.TransportDataListener;
import cgl.narada.transport.sslHttpBase.TransportImpl;
import cgl.narada.util.ByteUtilities;

/**
 * @author Hongbin Liu
 * @version 1.1
 */

public class HTTPSTransportImpl extends TransportImpl {

    final static Logger log = Logger.getLogger(HTTPSTransportImpl.class
        .getName());

    private int transactionNum = 0; // random?

    // Beyond this size means the message is corrupt.
    private static final int MAX_PAYLOAD_SIZE = 10000000;

    private Socket socket = null;

    // Indicated whether disconnect caused by io errors.
    private boolean ioerror = false;

    // private SocketFactory httpFactory = null;
    // private static Random s_rand = new Random(System.currentTimeMillis());

    private HTTPSReceiverThread m_receiverThread = new HTTPSReceiverThread(
                                                                           "httpsreceiver");

    private HTTPSSenderThread m_senderThread = new HTTPSSenderThread(
                                                                     "httpssender");

    private HTTPSDeliveryThread m_deliveryThread = null;

    private TransportDataListener m_listener = null;

    public static final int MAX_POST_SIZE = 20;

    public static final long PUSH_INTERVAL = 50;

    // transport factory server listening port; should make ONE variable
    public static final int DEFAULT_HTTPS_PORT = 20000;

    private int socketTimeout = 4000;

    class HTTPSSenderThread extends Thread {
        long lastCheck = System.currentTimeMillis();

        long lastStatusCheck = System.currentTimeMillis();

        boolean stopped = false;

        public HTTPSSenderThread(String name) {
            super(name);
        }

        public void run() {
            while (!stopped) {
                int numMessages = m_sendQueue.getNumMessages();
                long currentTime = System.currentTimeMillis();
                if (m_statusInterval != 0
                    && currentTime - lastStatusCheck > m_statusInterval) {
                    statusCheck();
                    log.debug("status check msg sent");
                    lastStatusCheck = currentTime;
                }
                if (numMessages >= MAX_POST_SIZE
                    || currentTime - lastCheck >= PUSH_INTERVAL) {
                    lastCheck = currentTime;
                    byte[] messages = m_sendQueue.removeAll();
                    // log.debug("send thread messages length " +
                    // messages.length);
                    if (messages.length == 0)
                        continue;
                    try {
                        sendDataImmediately(messages);
                    } catch (IOException ioe) {
                        ioe.printStackTrace();
                        handleSendException(ioe);
                    }
                }
                try {
                    Thread.sleep(100);
                } catch (InterruptedException ie) {
                    ie.printStackTrace();
                }
            }
        }

        public void statusCheck() {
            LinkMessageKeepAlive kam = new LinkMessageKeepAlive();
            byte[] bytes = kam.toBytes(); // type, channelId len, channelId,
            // payload len payload
            OutputStream out = null;
            try {
                out = socket.getOutputStream();
            } catch (IOException ioe) {
                ioe.printStackTrace();
            }
            int bodyLen = bytes.length;
            String headers = "POST http://" + m_host + ":" + m_port + "/"
                + (transactionNum++) + "HTTP/1.0\n"
                + "Proxy-Connection: Keep-Alive\n" + "Progma: no-cache\n"
                + "Content-Type: application/octet-stream\n"
                + "Content-Length: " + (bodyLen + 8) + "\r\n\n";
            ByteArrayOutputStream baos = new ByteArrayOutputStream();
            try {
                if (out != null) {
                    baos.write(headers.getBytes());
                    baos.write(ByteUtilities.getBytes(bodyLen + 4));
                    baos.write(ByteUtilities.getBytes(bodyLen));
                    baos.write(bytes);
                    out.write(baos.toByteArray());
                    log
                        .debug("status check sending http body "
                            + (bodyLen + 4));
                }
            } catch (IOException ioe) {
                ioe.printStackTrace();
            }
        }

        public void destroy() {
            if (!stopped) {
                stopped = true;
                interrupt();
            }
        }

    }

    class HTTPSReceiverThread extends Thread {
        boolean stopped = false;

        public HTTPSReceiverThread(String name) {
            super(name);
        }

        public void run() {
            while (!stopped) {
                try {
                    byte[] body = receiveDataImmediately();
                    if (body.length != 0)
                        m_receiveQueue.addAll(body);
                } catch (IOException ioe) {
                    ioe.printStackTrace();
                    handleReceiveException(ioe);
                }
            }
        }

        public void destroy() {
            if (!stopped) {
                stopped = true;
                interrupt();
            }
        }
    }

    /**
     * =======================================================================
     * Helper thread to take data off of the receive queue and invoke data
     * received on the listener.
     */
    class HTTPSDeliveryThread extends Thread {
        private boolean m_stopped = false;

        private Transport m_parent = null;

        HTTPSDeliveryThread(Transport parent) {
            m_parent = parent;
            setName("httsdelivery");
        }

        /**
         * 
         */
        public void run() {
            log.debug("delivery thread starting ... ");
            while (!m_stopped) {
                byte[] data = m_receiveQueue.removeData();
                if (data != null) {
                    m_listener.dataReceived(m_parent, data);
                } else {
                    if (!m_stopped) {
                        log
                            .warn("TransportImpl Delivery thread received null data.  Exiting");
                    }
                    destroy();
                }
            }
        }

        /**
         * 
         */
        public void destroy() {
            if (!m_stopped) {
                m_stopped = true;
                m_receiveQueue.dispose();
            }
        }
    }

    /**
     * -------------------------------------------------------------------
     */
    public HTTPSTransportImpl() {

    }

    /**
     * ------------------------------------------------------------------- Used
     * when an HTTPS transport end point is accepted rather than initiated.
     */
    protected HTTPSTransportImpl(Socket socket) {
        this.socket = socket;

        try {
            socket.setSoTimeout(socketTimeout);
        } catch (SocketException e1) {
            System.out.println("HTTPSTransportImpl: Problems setting timeout");
        }

        try {
            startThreads();
        } catch (IOException e) {
            log.error("Unable to start send/receive thread");
        }
    }

    /**
     * 
     */
    public void startThreads() throws IOException {
        m_receiverThread.start();
        m_senderThread.start();
    }

    /**
     * 
     */
    public void stopThreads() throws IOException {
        m_receiverThread.destroy();
        m_senderThread.destroy();
        if (m_deliveryThread != null)
            m_deliveryThread.destroy();
        if (m_listener != null)
            m_listener.transportClosed(this);
    }

    /**
     * 
     */
    public void setTransportDataListener(TransportDataListener l) {
        if (m_listener == null && l != null) {
            m_deliveryThread = new HTTPSDeliveryThread(this);
            m_deliveryThread.start();
        } else if (l == null && m_deliveryThread != null) {
            m_deliveryThread.destroy();
            m_deliveryThread = null;
        }
        m_listener = l;
    }

    /**
     * 
     */
    public boolean isConnected() {
        return socket != null;
    }

    /**
     * 
     */
    public boolean isInErrorState() {
        return ioerror;
    }

    /**
     * 
     */
    public String getType() {
        return HTTPSLinkFactory.CHANNEL_TYPE;
    }

    /**
     * 
     */
    public void connect() throws TransportException {
        if (m_host == null) {
            throw new TransportException("Unspecified endpoint m_host");
        }
        connect(m_host, m_port);
    }

    /**
     * 
     */
    public void connect(String host, int port) throws TransportException {
        String proxyHost = proxyInfo.getProxyHost();
        try {
            if (proxyHost != null) {
                proxyTunnel = new ProxyTunnel(proxyInfo);
                socket = proxyTunnel.doProxyConnect(host, port);
                log.debug("Connected to proxy @" + proxyHost + ":"
                    + proxyInfo.getProxyPort());
            } else {
                socket = new Socket(host, port);
                log.debug("Connected to server @: " + host + ":" + port);
            }

            socket.setSoLinger(false, 0);
            socket.setTcpNoDelay(true);

            startThreads();

        } catch (IOException e) {
            e.printStackTrace();
            throw new TransportException("IOError: " + e.getMessage());
        }
    }

    /**
     * 
     */
    public void connect(String host) throws TransportException {
        connect(host, DEFAULT_HTTPS_PORT);
    }

    /**
     * 
     */
    public synchronized void disconnect() throws IOException {
        if (socket != null) {
            stopThreads();
            socket.close();
            socket = null;
        }
    }

    /**
     * 
     */
    public boolean isSecure() {
        return false;
    }

    /**
     * 
     */
    public String toString() {
        if (m_host != null) {
            return "=>" + m_host + ":" + m_port;
        } else {
            return "<=" + socket.getInetAddress().getHostName() + ":"
                + socket.getLocalPort();
        }
    }

    /**
     * 
     */
    protected void throwBadConnectionException() throws IOException {
        if (ioerror) {
            throw new IOException("Connection in error state.  disconnected");
        } else {
            throw new IOException("Connection closed");
        }
    }

    /**
     * 
     */
    protected void sendDataImmediately(byte[] messages) throws IOException {
        if (socket == null)
            throwBadConnectionException();
        OutputStream out = socket.getOutputStream();
        ByteArrayOutputStream baos = new ByteArrayOutputStream();
        transactionNum++;
        int bodyLen = messages.length;
        String headers = "POST http://" + m_host + ":" + m_port + "/"
            + transactionNum + "HTTP/1.0\n" + "Proxy-Connection: Keep-Alive\n"
            + "Progma: no-cache\n" + "Content-Type: application/octet-stream\n"
            + "Content-Length: " + (bodyLen + 4) + "\r\n\n";
        baos.write(headers.getBytes());
        baos.write(ByteUtilities.getBytes(bodyLen));
        baos.write(messages);
        // log.debug("send immd");
        out.write(baos.toByteArray());
        out.flush();
        // log.debug("send immd. 2");
        int i = ByteUtilities.getInt(ByteUtilities.getBytes(bodyLen));
        // log.debug("send imme. sending "+ i);
    }

    /**
     * @return byte array that possibly include a series of messages with each
     *         led by 4 bytes of length. We usually don't need to use this since
     *         everything will be put in the receive queue.
     */
    protected byte[] receiveDataImmediately() throws IOException {
        if (socket == null)
            throwBadConnectionException();
        InputStream input = socket.getInputStream();

        int chars = 0;
        while (chars < 3) {
            int ch = input.read();
            if (ch < 0) {
                throw new IOException("Unexpected EOF from proxy");
            }
            if (ch == '\r' || ch == '\n') {
                chars++;
            } else if (ch != '\r' && ch != '\n') {
                chars = 0;
            }
        }
        byte[] lenBytes = new byte[4];
        input.read(lenBytes);
        int len = ByteUtilities.getInt(lenBytes, 0);
        // log.debug("receive imme. " + len);
        if (len > MAX_PAYLOAD_SIZE)
            throw new IOException("Bad size or payload too large: " + len);
        byte[] body = readBytes(input, len);
        if (handleTransportSystemMessage(lenBytes)) {
            return new byte[0];
        }
        // log.debug("receive imme. received "+ body.length);
        int cnt = 0;
        for (int pos = 0; pos < body.length; pos += 4) {
            int slen = ByteUtilities.getInt(body, pos);
            // log.debug("receive single message len " + slen);
            pos += slen;
            cnt++;
        }
        // log.debug("receive imme count = " + cnt);
        return body;
    }

    /**
     * 
     */
    protected void handleReceiveException(Exception e) {
        // Handle by closing the connection.
        try {
            ioerror = true;
            disconnect();
        } catch (IOException ex) {
            ex.printStackTrace();
        }
    }

    /**
     * 
     */
    protected void handleSendException(Exception e) {
        // Handle by closing the connection.
        try {
            ioerror = true;
            disconnect();
        } catch (IOException ex) {
            ex.printStackTrace();
        }
    }

    /**
     * 
     */
    private byte[] readBytes(InputStream input, int size) throws IOException {
        byte[] buf = new byte[size];

        int remainingSize = size;
        int readSize = 0;
        while ((readSize = input.read(buf, (size - remainingSize),
                                      remainingSize)) < remainingSize) {
            if (readSize < 0) {
                throw new IOException(
                                      "Stream terminated before chunk can be fully read[size="
                                          + size + "]");
            }
            remainingSize -= readSize;
        }
        return buf;
    }
}
